package com.qiandw.common.cassandra

import com.datastax.driver.core.ProtocolOptions.Compression
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy
import com.datastax.driver.core.{Cluster, ConsistencyLevel, QueryOptions}
import org.apache.commons.lang.StringUtils
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.{DisposableBean, InitializingBean}
import org.springframework.stereotype.Component

import scala.beans.BeanProperty

@Component
class ClusterFactoryImpl extends ClusterFactory with InitializingBean with DisposableBean {

  @Autowired
  var cfg :CassandraConfig = _

  private var cluster: Cluster = _

  @BeanProperty
  var contactNodes: String = _

  @BeanProperty
  var port: Integer = _

  @BeanProperty
  var username: String = _

  @BeanProperty
  var password: String = _

  @BeanProperty
  var datacenter: String = _

  @BeanProperty
  var compression: Boolean = _


  override def getCluster = {
    if (cluster == null || cluster.isClosed)
      createCluster()
    cluster
  }

  override def afterPropertiesSet() = {

    contactNodes = cfg.nodes
    port = cfg.port
    username = cfg.user
    password = cfg.password
    datacenter = cfg.datacenter
    compression = cfg.compression

    cluster = getBuilder.build
    cluster.init
  }

  override def destroy() = {
    cluster.close()
  }

  private def createCluster(): Unit = {
    cluster = getBuilder.build
    cluster.init
  }

  private def getBuilder = {
    val b = Cluster.builder
    if (StringUtils.isEmpty(contactNodes)) throw new Exception("No contact point defined.")
    val cps = contactNodes.split(",")
    for (cp <- cps) {
      b.addContactPoint(cp.trim)
    }
    if (port > 0) b.withPort(port)
    if (StringUtils.isNotEmpty(username) && StringUtils.isNotEmpty(password))
      b.withCredentials(username, password)
    if (StringUtils.isNotEmpty(datacenter))
      b.withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder.withLocalDc(datacenter).build)
    if (compression) b.withCompression(Compression.LZ4)
    //设置默认的语句属性
    val qos = new QueryOptions
    qos.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM)
    b.withQueryOptions(qos)
    b
  }
}
